package job.index.indegree_distribution;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import mapred.base.TextMapper;

import org.apache.hadoop.io.Text;

public class Indegree_Distribution_Mapper extends TextMapper {

    private static List<Long> list = new ArrayList<Long>();

    @Override
    public void map(Text key, Text value) throws IOException {
        int pos = Collections.binarySearch(list, Long.parseLong(value
                .toString()));
        if (pos < 0) {
            pos = -pos - 1;
        }
        this.incrCounter("Counter", "(" + list.get(pos - 1) + ", "
                + list.get(pos) + "]", 1L);
    }

    @Override
    public void setUp() {
        String[] t = this.getConfig("point").split(",");
        for (String s : t) {
            list.add(Long.valueOf(s));
        }
        list.add(-1L);
        list.add(Long.MAX_VALUE);
        Collections.sort(list);
    }

    @Override
    public void tearDown() throws IOException {
    }

}
